package com.gmscosmo.janus.queue;

import com.gmscosmo.janus.config.JanusConfig;
import com.gmscosmo.janus.util.JanusLogUtil;
import com.rabbitmq.client.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Component
public class RabbitMqUtil {
    JanusConfig janusConfig;
    @Autowired
    JanusLogUtil janusLogUtil;

    private static ConnectionFactory factory;
    private static Connection connection;
    private static ConcurrentHashMap<String,Channel> channels;

    @Autowired
    public void setJanusConfig(JanusConfig janusConfig) {
        this.janusConfig = janusConfig;
        String host = janusConfig.getRabbitmq_host();
        int port = janusConfig.getRabbitmq_port();
        String userName = janusConfig.getRabbitmq_username();
        String password = janusConfig.getRabbitmq_password();
        String virtualHost = janusConfig.getRabbitmq_virtualHost();
        factory = this.initConnectionFactory(host, port, userName, password, virtualHost);
        // 创建与RabbitMQ服务器的TCP连接
        try {
            connection = connection == null ? factory.newConnection() : connection;
            channels = new ConcurrentHashMap<>(16);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化rabbitMq服务配置
     *
     * @param host        主机
     * @param port        端口
     * @param userName    用户名
     * @param password    密码
     * @param virtualHost 虚拟主机
     * @return
     */
    private ConnectionFactory initConnectionFactory(String host, int port, String userName, String password, String virtualHost) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(userName);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
        return factory;
    }

    /**
     * 绑定队列
     *
     * @param exchangeName 交换机名
     * @param queneName    队列名
     * @param routingKey   路由KEY
     * @param type         消息模式：FANOUT|TOPIC|DIRECT
     * @param durable      是否持久化
     * @param autoDelete   是否自动删除队列
     * @throws IOException
     */
    private void queueBind(String exchangeName, String queneName, String routingKey, BuiltinExchangeType type, boolean durable, boolean autoDelete) throws IOException {
        // 声明交换机类型：交换机，类型，持久化
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchangeName, type, durable);
        if (queneName != null) {
            if (type != BuiltinExchangeType.DIRECT) {
                // 声明默认的队列：队列，持久化，声明独占队列（仅限于此连接），自动删除队列，队列的其他属性
                channel.queueDeclare(queneName, durable, false, autoDelete, null);
            }
            // 将队列与交换机绑定
            channel.queueBind(queneName, exchangeName, routingKey);
        }
        channels.put(queneName,channel);
    }

    /**
     * 发送消息
     *
     * @param queneName    队列名
     * @param routingKey   路由KEY
     * @param type         消息模式：FANOUT|TOPIC|DIRECT
     * @param msg          消息
     * @return
     */
    public boolean sendMq(String queneName, String routingKey, BuiltinExchangeType type, String msg) {
        try {
            if(!channels.containsKey(queneName)) {
                this.queueBind(janusConfig.getRabbitmq_exchange(), queneName, routingKey, type, false, false);
            }
            channels.get(queneName).basicPublish(janusConfig.getRabbitmq_exchange(), routingKey, null, msg.getBytes());
            janusLogUtil.log(queneName + "生产数据:"+ msg,this.getClass(),JanusLogUtil.DEBUG);
        } catch (Exception e) {
            janusLogUtil.log("error:"+ e,this.getClass(), JanusLogUtil.ERROR);
            return false;
        }
        return true;
    }

    public void ack(String routing_Key ,long tag) {
        try {
            channels.get(routing_Key).basicAck(tag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void nack(String routing_Key,long tag) {
        try {
            channels.get(routing_Key).basicNack(tag, false, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 拉取队列消息
     *
     * @param queneName       队列名
     * @param routingKey      路由KEY
     * @param type            消息模式：FANOUT|TOPIC|DIRECT
     * @param headerInterface 回调实现
     * @throws IOException
     */
    public void pullMq(String queneName, String routingKey, BuiltinExchangeType type, HeaderInterface headerInterface) throws IOException {
        String exchangeName = janusConfig.getRabbitmq_exchange();
        this.queueBind(exchangeName, queneName, routingKey, type, false, false);
        // 创建接收客户端，当有消息，则直接回调handleDelivery方法
        Channel channel = channels.get(queneName);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                headerInterface.execute(consumerTag, envelope.getRoutingKey(),envelope.getDeliveryTag(), body);
            }
        };
        // channel绑定队列、消费者，autoAck为true表示一旦收到消息则自动回复确认消息
        channel.basicConsume(queneName, false, consumer);
    }

    /**
     * 关闭连接通道
     *
     * @throws IOException
     * @throws TimeoutException
     */
    @PreDestroy
    public void close() throws IOException, TimeoutException {
        if (channels.size() != 0) {
            for(Channel channel:channels.values()){
                janusLogUtil.log("关闭管道"+channel,this.getClass(),JanusLogUtil.DEBUG);
                channel.close();
                channel = null;
            }
        }
        if (connection != null) {
            janusLogUtil.log("关闭tcp连接"+connection,this.getClass(),JanusLogUtil.DEBUG);
            connection.close();
            connection = null;
        }
        factory = null;
    }

    /**
     * 函数式回调接口
     */
    @FunctionalInterface
    interface HeaderInterface {
        void execute(String consumerTag,String routingKey, long deliveryTag, byte[] body) throws IOException;
    }

}
